package com.nlp.zk;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

public class CuratorDemo {
	String connectString = "119.23.250.150:2181,119.23.250.150:2182,119.23.250.150:2183";
	
	private CuratorFramework framework;
	
	int i = 0;
	public CuratorDemo() {
		CuratorFramework framework = CuratorFrameworkFactory
				.builder()
				.sessionTimeoutMs(40000)
				.retryPolicy(new ExponentialBackoffRetry(10000,3))
				.connectString(connectString)
				.namespace("curator")
				.build();
		framework.start();
		this.framework = framework;
	}
	
	/**
	 * 创建node
	 * @param path
	 * @param data
	 */
	public void create(String path, String data) {
		try {
			String str = framework.create().creatingParentsIfNeeded()
			.withMode(CreateMode.PERSISTENT)
			.forPath(path,data.getBytes());
			System.out.println(str);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 得到node数据
	 * @param path
	 * @param stat
	 * @return
	 */
	public String getData(String path, Stat stat) {
		byte[] data = null;
		try {
			data = framework.getData().storingStatIn(stat).forPath(path);
			return new String(data);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	/**
	 * 修改node
	 * @param path
	 * @param stat
	 * @param data
	 * @return
	 */
	public Stat setData(String path, Stat stat, String data) {
		try {
			return stat = framework.setData().withVersion(stat.getAversion()).forPath(path, data.getBytes());
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	/**
	 * 删除node
	 * @param path
	 * @return
	 */
	public boolean delete(String path) {
		//删除
		try {
			framework.delete().deletingChildrenIfNeeded().forPath(path);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return false;
	}
	
	/**
	 * 节点的增删改监听
	 * @param path
	 * @throws Exception
	 */
	public void watcherNode(String path) throws Exception {
		NodeCache nodeCache = new NodeCache(framework, path, false);
		NodeCacheListener listener = new NodeCacheListener() {
			
			@Override
			public void nodeChanged() throws Exception {
				System.out.println("recevive event:"+nodeCache.getCurrentData().getPath());
				
			}
		};
		
		nodeCache.getListenable().addListener(listener);
		nodeCache.start();
	}
	
	/**
	 * 自己点的增删改监听
	 * @param path
	 * @throws Exception
	 */
	public void watcherChildChange(String path) throws Exception {
		PathChildrenCache childrenCache = new PathChildrenCache(framework, path, true);
		PathChildrenCacheListener listener = new PathChildrenCacheListener() {
			
			@Override
			public void childEvent(CuratorFramework framework, PathChildrenCacheEvent event) throws Exception {
				System.out.println("Receive Event:"+event.getType());
				
			}
		};
		
		childrenCache.getListenable().addListener(listener);
		childrenCache.start();
	}
	
	/**
	 * node任何变化的监听
	 * @param path
	 */
	public void watcherTree(String path) {
		TreeCache treeCache = new TreeCache(framework, path);
		TreeCacheListener listener = new TreeCacheListener() {
			
			@Override
			public void childEvent(CuratorFramework framework, TreeCacheEvent event) throws Exception {
				System.out.println("Receive Event:"+event.getType());				
			}
		};
		treeCache.getListenable().addListener(listener);
		try {
			treeCache.start();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public void close() {
		framework.close();
	}
	
	
	public void lock(String path) {
		InterProcessMutex mutex = new InterProcessMutex(framework, path);
		try {
			//加锁
			mutex.acquire();
			System.out.println("线程【"+Thread.currentThread().getName()+"】获取到锁");
			i++;
			//System.out.println(i);
		} catch (Exception e) {
			e.printStackTrace();
		}finally {
			try {
				//释放锁
				mutex.release();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	
	public static void main(String[] args) throws Exception {

		CuratorDemo demo = new CuratorDemo();
		String path = "/abcft";
		String data = "222";
		//demo.create(path, data);
		
		//demo.close();
		//demo.watcherNode(path);
		//demo.watcherChildChange(path);
		

		ExecutorService es = Executors.newFixedThreadPool(10);
		CountDownLatch latch = new CountDownLatch(10);
		for (int i = 0; i < 10; i++) {
			String name = "thread"+i;
			es.execute(new Thread(name) {
				@Override
				public void run() {
					demo.lock(path);
					System.out.println("线程【"+Thread.currentThread().getName()+"】释放锁");
					latch.countDown();
				}
			});
		}
		latch.await();
		System.out.println(demo.i);
		es.shutdown();
//		demo.watcherTree(path);
//		System.in.read();
		
		
	}

}
